package io.xxx.omni.oss

import org.apache.pulsar.client.api.PulsarClient
import org.apache.pulsar.client.api.Schema
import org.apache.pulsar.client.api.SubscriptionInitialPosition
import java.util.*
import java.util.concurrent.TimeUnit

class PulsarConsumeTests {
}

fun main() {
    val client = PulsarClient.builder()
        .serviceUrl("pulsar://10.48.137.117:6650,10.48.137.118:6650,10.48.137.119:6650")
        .build()

    val random = Random()

    val consumer = client.newConsumer(Schema.STRING)
        .negativeAckRedeliveryDelay(5000, TimeUnit.SECONDS)
//        .topicsPattern(Pattern.compile("persistent://public/default/test"))
        .topic("persistent://public/default/document-aikucun")
        .subscriptionName("oss-subscription")
        .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
        .messageListener { consumer, msg ->
            try {
                System.err.println("收到消息: ${String(msg.data).substring(0, 20)} : ${msg.messageId} ")
//                1 / random.nextInt(3)
                consumer.acknowledge(msg)
            } catch (e: Exception) {
                consumer.negativeAcknowledge(msg)
                throw e
            }
        }
        .subscribe()

//    while (true) {
//        // Wait for a message
//        val msg = consumer.receive()
//        try {
//            // Do something with the message
//            System.err.println("收到消息: " + String(msg.data))
//
//            // Acknowledge the message so that it can be deleted by the message broker
//            consumer.acknowledge(msg)
//        } catch (e: Exception) {
//            // Message failed to process, redeliver later
//            consumer.negativeAcknowledge(msg)
//        }
//    }
}